package com.deep.nacos.config;

import com.alibaba.cloud.nacos.NacosConfigManager;
import com.alibaba.cloud.nacos.NacosConfigProperties;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.nacos.api.config.listener.Listener;
import lombok.Data;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.Constructor;

import javax.annotation.Resource;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;


/**
 * 动态线程池
 */
@Configuration
@Data
public class MyDynamicThreadPool implements InitializingBean {

    @Value("${thread.pool.corePoolSize}")
    private int corePoolSize;
    @Value("${thread.pool.maxPoolSize}")
    private int maxPoolSize;
    @Value("${thread.pool.queueCapacity}")
    private int queueCapacity;
    @Value("${thread.pool.keepAliveSeconds}")
    private int keepAliveSeconds;

    private static ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    private NacosConfigManager nacosConfigManager;

    @Resource
    private NacosConfigProperties nacosConfigProperties;


    @Override
    public void afterPropertiesSet() throws Exception {
        // 初始化线程池
        threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
        threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
        threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
        threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveSeconds);
        threadPoolTaskExecutor.setThreadNamePrefix(System.currentTimeMillis() + "_");
        threadPoolTaskExecutor.setRejectedExecutionHandler(
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.out.println("队列已满，抛弃任务");
                    }
                });
        threadPoolTaskExecutor.initialize();

        // 增加nacos配置监听
        nacosConfigManager.getConfigService().addListener("thread-pool.yml", nacosConfigProperties.getGroup(),
                new Listener() {
                    @Override
                    public Executor getExecutor() {
                        return null;
                    }

                    @Override
                    public void receiveConfigInfo(String configInfo) {
                        System.out.println("动态修改前-->");
                        print();
                        Yaml yaml = new Yaml(new Constructor(JSONObject.class));
                        InputStream inputStream = new ByteArrayInputStream(configInfo.getBytes());
                        JSONObject data = yaml.load(inputStream);
                        JSONObject pool = data.getJSONObject("thread").getJSONObject("pool");
                        threadPoolTaskExecutor.setCorePoolSize(pool.getInteger("corePoolSize"));
                        threadPoolTaskExecutor.setMaxPoolSize(pool.getInteger("maxPoolSize"));
                        System.out.println("动态修改后-->");
                        print();
                    }
                });

    }

    //执行任务
    public void execute(Runnable runnable) {
        threadPoolTaskExecutor.execute(runnable);
    }

    public void print() {
        System.out.println("核心线程数：" + threadPoolTaskExecutor.getThreadPoolExecutor().getCorePoolSize() + " " + "最大线程数：" + threadPoolTaskExecutor.getThreadPoolExecutor().getMaximumPoolSize()
                + " " + "阻塞队列数：" + threadPoolTaskExecutor.getThreadPoolExecutor().getQueue().size() + "/" + queueCapacity + " " + "活跃线程数：" + threadPoolTaskExecutor.getThreadPoolExecutor().getActiveCount());
    }
}